package server_agent

import (
	"lark/proto"
	"lark/util"
	"net"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/pkg/errors"
	"lark/conf"
)

type ServerInterface interface {
	Run()
	TryClose() bool
	LastKeepAliveTime() time.Time
	KeepAlive()
	DelClientConn(string)
	DelServerConn(string)
	TaskChan() chan *TaskReqMsg
	ServiceName() string
}

const (
	RegisterKindPHP  = "php"
	RegisterKindJava = "java"
)

type Server struct {
	sync.Mutex
	registInfo  ServiceRegistInfo
	id          string
	lis         net.Listener
	taskChan    chan *TaskReqMsg
	clientCount uint64
	clientConns map[string]ClientConnInterface //key 是来源 IP
	serverConns map[string]*serverConn         //key 是编号
	status      uint32
}

type TaskReqMsg struct {
	Client    ClientConnInterface
	Req       *proto.TaskReq
	Raw       bool //是否只返回原始数据，老的直连客户端需要设置为 true
	ClientIP  string
	CreatedAt time.Time
}

func NewService(registInfo ServiceRegistInfo) (ServerInterface, error) {
	if registInfo.Kind == "" {
		registInfo.Kind = RegisterKindPHP
	}
	registInfo.KeepAliveTime = time.Now()

	p := &Server{
		registInfo:  registInfo,
		taskChan:    make(chan *TaskReqMsg, conf.ServerAgentRequestBacklog),
		clientConns: make(map[string]ClientConnInterface),
		serverConns: make(map[string]*serverConn),
		id:          registInfo.String(),
	}

	l, err := net.Listen("tcp", "0.0.0.0:"+strconv.Itoa(p.registInfo.Port))
	if err != nil {
		return nil, errors.WithStack(err)
	}
	p.lis = l

	for i := 0; i < p.registInfo.WorkNum; i++ {
		id := strings.Join([]string{p.id, strconv.Itoa(i)}, "-")
		s := newServerConn(id, p.taskChan, p.registInfo.CgiPass, p.registInfo.Kind, p)
		logger.Infof("new server conn. id:%s", id)
		p.serverConns[id] = s
		go s.run()
	}
	p.status = runningFlag

	return p, nil
}

func (p *Server) Run() {
	defer util.PrintPanicStack()

	for {
		conn, err := p.lis.Accept()
		if err != nil {
			if p.status == closedFlag {
				logger.Debugf("[%s] break accept by close", p.id)
				return
			}

			if ne, ok := err.(net.Error); ok && ne.Temporary() {
				time.Sleep(time.Millisecond * 100)
			} else {
				time.Sleep(time.Second)
			}

			logger.Errorf("[%s] accept error. %s", p.id, err)
			continue
		}

		p.clientCount++
		id := "client-" + conn.RemoteAddr().String() + "-" + strconv.FormatUint(p.clientCount, 10)
		logger.Debugf("[%s] new client conn. id:%s", p.id, id)

		client := NewClient(id, conn, p)

		p.Lock()
		p.clientConns[id] = client
		p.Unlock()

		go client.Run()
	}
}

func (p *Server) serviceRegistInfo() ServiceRegistInfo {
	return p.registInfo
}

func (p *Server) DelClientConn(id string) {
	p.Lock()
	delete(p.clientConns, id)
	p.Unlock()
}

func (p *Server) DelServerConn(id string) {
	p.Lock()
	delete(p.serverConns, id)
	p.Unlock()
}

func (p *Server) close() {
	if atomic.CompareAndSwapUint32(&p.status, runningFlag, closedFlag) {
		p.lis.Close()
		for _, c := range p.clientConns {
			c.Close()
		}

		close(p.taskChan)
	}
}

func (p *Server) TryClose() bool {
	p.Lock()
	defer p.Unlock()

	if len(p.clientConns) == 0 {
		p.close()
		return true
	}

	return false
}

func (p *Server) LastKeepAliveTime() time.Time {
	p.Lock()
	defer p.Unlock()

	return p.registInfo.KeepAliveTime
}

func (p *Server) KeepAlive() {
	p.Lock()
	defer p.Unlock()

	p.registInfo.KeepAliveTime = time.Now()
}

func (p *Server) TaskChan() chan *TaskReqMsg {
	return p.taskChan
}

func (p *Server) ServiceName() string {
	return p.registInfo.ServiceName
}
